package com.deng.eventflow.core;

import com.deng.eventflow.core.constant.NextStatus;
import com.deng.eventflow.core.exception.ProcessorException;
import com.deng.eventflow.core.exception.PullableException;
import com.deng.eventflow.core.exception.PushableException;

import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.Future;

public abstract class UniformProcessor extends SynchronousProcessor {

    protected transient Object[] m_outputArray;

    public UniformProcessor(int in_arity, int out_arity, /*@ null @*/ Context c) {
        super(in_arity, out_arity);
        m_outputArray = new Object[out_arity];

    }

    public UniformProcessor(int in_arity, int out_arity) {
        this(in_arity, out_arity, null);
    }

    @Override
    protected final boolean compute(Object[] inputs, Queue<Object[]> outputs)
    {
        boolean b = compute(inputs, m_outputArray);
        outputs.add(m_outputArray);
        return b;
    }

    protected abstract boolean compute(Object[] inputs, Object[] outputs);


    protected boolean onEndOfTrace(Object[] outputs)
    {
        return false;
    }

    @Override
    protected final boolean onEndOfTrace(Queue<Object[]> outputs)
    {
        Object[] outs = new Object[getOutputArity()];
        boolean b = onEndOfTrace(outs);
        if (b)
        {
            outputs.add(outs);
        }
        return b;
    }

    @Override
    public Pullable getPullableOutput(int index)
    {
        if (m_outputPullables[index] == null)
        {
            if (m_inputArity == 1 && m_outputArity == 1)
            {
                m_outputPullables[index] = new UnaryPullable();
            }
            else
            {
                m_outputPullables[index] = new OutputPullable(index);
            }
        }
        return m_outputPullables[index];
    }

    @Override
    public Pushable getPushableInput(int index)
    {
        if (m_inputPushables[index] == null)
        {
            if (m_inputArity == 1 && m_outputArity == 1)
            {
                m_inputPushables[index] = new UnaryPushable();
            }
            else
            {
                m_inputPushables[index] = new InputPushable(index);
            }
        }
        return m_inputPushables[index];
    }



    public class UnaryPushable implements Pushable   {
        @Override
        public synchronized Pushable push(Object o)
        {
            try
            {
                compute(new Object[] { o }, m_outputArray);
            }
            catch (ProcessorException e)
            {
                throw new PushableException(e);
            }
            if (m_outputPushables[0] == null)
            {
                throw new PushableException(
                        "Output 0 of processor " + getProcessor() + " is connected to nothing");
            }
            m_outputPushables[0].push(m_outputArray[0]);
            return this;
        }

        @Override
        public synchronized Future<Pushable> pushFast(Object o)
        {
            push(o);
            return Pushable.NULL_FUTURE;
        }

        @Override
        public synchronized void notifyEndOfTrace() throws PushableException
        {
            m_hasBeenNotifiedOfEndOfTrace[getPosition()] = true;
            if (!allNotifiedEndOfTrace())
            {
                return;
            }
            boolean b;
            try
            {
                b = onEndOfTrace(m_outputArray);
            }
            catch (ProcessorException e)
            {
                throw new PushableException(e);
            }
            if (b)
            {
                m_outputPushables[0].push(m_outputArray[0]);
            }
            for (int i = 0; i < m_outputArity; i++)
            {
                m_outputPushables[i].notifyEndOfTrace();
            }
        }

        @Override
        public UniformProcessor getProcessor()
        {
            return UniformProcessor.this;
        }

        @Override
        public int getPosition()
        {
            return 0;
        }
    }


    public class UnaryPullable implements Pullable
    {

        @Override
        public Iterator<Object> iterator()
        {
            return this;
        }

        @Override
        public void remove()
        {
            throw new UnsupportedOperationException();
        }

        @Override
        public Object pullSoft()
        {
            if (!m_inputQueues[0].isEmpty())
            {
                return m_inputQueues[0].remove();
            }
            Object o = m_inputPullables[0].pullSoft();
            try
            {
                if (o == null || !compute(new Object[] { o }, m_outputArray))
                {
                    return null;
                }
            }
            catch (ProcessorException e)
            {
                throw new PullableException(e);
            }
            return m_outputArray[0];
        }

        @Override
        public Object pull()
        {
            if (!m_inputQueues[0].isEmpty())
            {
                return m_inputQueues[0].remove();
            }
            if (m_inputPullables[0] == null)
            {
                throw new PullableException("Input 0 of this processor is connected to nothing",
                        getProcessor());
            }
            Object o = m_inputPullables[0].pull();
            try
            {
                if (o == null || !compute(new Object[] { o }, m_outputArray))
                {
                    throw new NoSuchElementException();
                }
            }
            catch (ProcessorException e)
            {
                throw new PullableException(e);
            }
            return m_outputArray[0];
        }

        @Override
        @SuppressWarnings("squid:S2272") // since() pull throws the exception
        public Object next()
        {
            return pull();
        }

        @Override
        public NextStatus hasNextSoft()
        {
            if (!m_inputQueues[0].isEmpty())
            {
                // Since we are a uniform processor, we know that the
                // existence of an input will generate an output
                return NextStatus.YES;
            }
            else
            {
                if (m_inputPullables[0] == null)
                {
                    throw new PullableException("Input 0 of this processor is connected to nothing",
                            getProcessor());
                }
                return m_inputPullables[0].hasNextSoft();
            }
        }

        @Override
        public boolean hasNext()
        {
            if (!m_inputQueues[0].isEmpty())
            {
                // Since we are a uniform processor, we know that the
                // existence of an input will generate an output
                return true;
            }
            else
            {
                if (m_inputPullables[0] == null)
                {
                    throw new PullableException("Input 0 of this processor is connected to nothing",
                            getProcessor());
                }
                return m_inputPullables[0].hasNext();
            }
        }

        @Override
        public Processor getProcessor()
        {
            return UniformProcessor.this;
        }

        @Override
        public int getPosition()
        {
            return 0;
        }

        @Override
        public void start()
        {
            UniformProcessor.this.start();
        }

        @Override
        public void stop()
        {
            UniformProcessor.this.stop();
        }

        @Override
        public void dispose()
        {
            // Nothing to do
        }

    }


}
